package com.nota.flink.cdc.main;


import cn.hutool.extra.spring.SpringUtil;
import com.alibaba.fastjson.JSON;
import com.nota.flink.cdc.consumer.RedisStreamConsumer;
import com.nota.flink.cdc.pojo.BinlogInfo;
import com.nota.flink.cdc.pojo.DataChangeInfo;
import com.nota.flink.cdc.pojo.constants.RedisStreamConstants;
import com.nota.flink.cdc.utils.CommonUtil;
import com.nota.flink.cdc.utils.RedisUtil;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.springframework.stereotype.Component;

@Slf4j
@Component
public class DataChangeSink implements SinkFunction<DataChangeInfo> {


    @Override
    public void invoke(DataChangeInfo dataChangeInfo, Context context) {
        try {
            RedisUtil redisUtil = SpringUtil.getBean(RedisUtil.class);
            // 变更类型： 0 初始化 1新增 2修改 3删除 4导致源中的现有表被截断的操作
            if (!SpringUtil.getBean(RedisStreamConsumer.class).consumeData(dataChangeInfo)) {
                SpringUtil.getBean(CommonUtil.class).errorHandler("消息消费失败,进入死信队列", JSON.toJSONString(dataChangeInfo),
                        "消息消费失败,当前消息id[{}]," + "消息实体[{}]", null);
            }
            //不管有没有失败,都把当前的 log 文件 和 位置放入到 redis
            setBinlogInfoToRedis(dataChangeInfo, redisUtil);
            log.info("读取到mysql数据第{}条数据,时间{}", increment(),
                    dataChangeInfo.getOperatorTime());
        } catch (Exception e) {
            SpringUtil.getBean(CommonUtil.class).errorHandler("消息invok失败,进入死信队列", JSON.toJSONString(dataChangeInfo),
                    "消息消费失败,当前消息id[{}]," + "消息实体[{}]", e);
        }

    }

    private void setBinlogInfoToRedis(DataChangeInfo dataChangeInfo, RedisUtil redisUtil) {
        try {
            BinlogInfo binlogInfo = new BinlogInfo();
            binlogInfo.setOperatorTime(dataChangeInfo.getOperatorTime());
            binlogInfo.setFilePos(dataChangeInfo.getFilePos());
            binlogInfo.setFileName(dataChangeInfo.getFileName());
            redisUtil.setCacheObject(RedisStreamConstants.BINLOG, dataChangeInfo);
        } catch (Exception e) {
            log.error("设置位置信息到redis失败", e);
        }
    }

    public int increment() {
        return MysqlEventListener.COUNT.getAndIncrement();
    }


}